package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.bean.TrafficHomeDetailPageViewBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.DorisUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * ClassName: DwsTrafficHomeDetailPageViewWindow
 * Package: com.atguigu.edu.realtime.app.dws
 * Description:
 *
 * @Author Mr.2
 * @Create 2023/9/11 15:41
 * @Version 1.0
 */
public class DwsTrafficHomeDetailPageViewWindow {
    public static void main(String[] args) {
        // TODO 1. 指定流处理环境
        // 1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(4);

        // TODO 2. 检查点相关的设置
        // 2.1 开启检查点
        env.enableCheckpointing(10000L, CheckpointingMode.AT_LEAST_ONCE);
        // 2.2 设置 检查点超时时间
        // 2.3 设置 Job取消后，检查点是否保留
        // 2.4 设置 两个检查点之间最小时间间隔
        // 2.5 设置 重启策略 -- 可以采用故障率重启策略
        // 2.6 设置 状态后端(即:状态的保存位置)
        // 2.7 设置 系统的操作用户

        // TODO 3. From Kafka topic 读取数据
        // 3.1 声明消费的主题 以及 消费者组
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_home_detail_group";
        // 3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        // 3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS =
                env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");

        // TODO 4. 对流中数据进行数据类型转换 JSONString -> JSONObject
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);

        // TODO 5. 过滤出首页(home)以及详情页(course_detail)的日志
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(
                new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String pageId = jsonObject.getJSONObject("page").getString("page_id");
                        return "home".equals(pageId) || "course_detail".equals(pageId);
                    }
                }
        );
        // For test output
//        filterDS.print("filter->");

        // TODO 6. 指定watermark以及提取事件时间(event time)字段
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = filterDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                }
                        )
        );

        // TODO 7. 按照设备(mid)进行分组
        KeyedStream<JSONObject, String> keyedDS =
                withWatermarkDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        // TODO 8. 使用Flink的状态编程, 判断是否为首页(home)以及详情页(course_detail) JSONObject -> 实体类对象
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> processedDS = keyedDS.process(
                new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {

                    // 声明状态
                    private ValueState<String> homeLastVisitDateState;
                    private ValueState<String> detailLastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 状态初始化
                        ValueStateDescriptor<String> homeValueStateDescriptor
                                = new ValueStateDescriptor<>("homeLastVisitDateState", String.class);
                        // 设置状态保存时间, 状态保持一天
                        homeValueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                        homeLastVisitDateState = getRuntimeContext().getState(homeValueStateDescriptor);

                        ValueStateDescriptor<String> detailValueStateDescriptor
                                = new ValueStateDescriptor<>("detailLastVisitDateState", String.class);
                        // 设置状态保存时间, 状态保持一天
                        detailValueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                        detailLastVisitDateState = getRuntimeContext().getState(detailValueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                        // 获取 页面id
                        String pageId = jsonObj.getJSONObject("page").getString("page_id");

                        // 获取 当前访问日期
                        Long ts = jsonObj.getLong("ts");
                        String curVisitDate = DateFormatUtil.toDate(ts);

                        // 统计指标 初始化
                        long homeUvCt = 0L;
                        long detailUvCt = 0L;

                        if ("home".equals(pageId)) {
                            // 获取当前设备上次访问首页的日期
                            String homeLastVisitDate = homeLastVisitDateState.value();
                            if (StringUtils.isEmpty(homeLastVisitDate) || !homeLastVisitDate.equals(curVisitDate)) {
                                // 说明 是首页 新访客
                                homeUvCt = 1L;
                                // 更新到状态中
                                homeLastVisitDateState.update(curVisitDate);
                            }
                        } else {
                            // 获取当前设备上次访问详情页的日期
                            String detailLastVisitDate = detailLastVisitDateState.value();
                            if (StringUtils.isEmpty(detailLastVisitDate) || !detailLastVisitDate.equals(curVisitDate)) {
                                // 说明 是 详情页 新访客
                                detailUvCt = 1L;
                                // 更新到状态中去
                                detailLastVisitDateState.update(curVisitDate);
                            }
                        }

                        // 判断是否为首页或者详情页的独立访客
                        if (homeUvCt != 0L || detailUvCt != 0L) {
                            TrafficHomeDetailPageViewBean viewBean
                                    = new TrafficHomeDetailPageViewBean("", "", "", homeUvCt, detailUvCt);
                            // 向 下游传递
                            out.collect(viewBean);
                        }

                    }
                }
        );

        // TODO 9. 开窗
        AllWindowedStream<TrafficHomeDetailPageViewBean, TimeWindow> windowalledDS
                = processedDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));

        // TODO 10. 聚合
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> reducedDS = windowalledDS.reduce(
                new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                    @Override
                    public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1, TrafficHomeDetailPageViewBean value2) throws Exception {
                        value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                        value1.setCourseDetailUvCt(value1.getCourseDetailUvCt() + value2.getCourseDetailUvCt());
                        return value1;
                    }
                },
                /*
                 *  对 聚合后的数据,进行补充时间相关属性(window_start, window_end、cur_date 即Doris分区字段)
                 * <IN, OUT, WINDOW>
                 * */
                new ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<TrafficHomeDetailPageViewBean> input, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {

                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String curDate = DateFormatUtil.toDate(context.window().getStart());

                        for (TrafficHomeDetailPageViewBean viewBean : input) {
                            viewBean.setStt(stt);
                            viewBean.setEdt(edt);
                            viewBean.setCurDate(curDate);

                            // 向下游发送
                            out.collect(viewBean);
                        }
                    }
                }
        );
        // For test output
        reducedDS.print("reduced ->");

        // TODO 11. 将聚合结果写到Doris
        reducedDS
                .map(new BeanToJsonStrFunction<TrafficHomeDetailPageViewBean>())
                .sinkTo(DorisUtil.getDorisSink("dws_traffic_home_detail_page_view_window"));

        // 执行 环境
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
